package com.clei.utils.io;

import com.clei.utils.PrintUtil;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * 对DataMap的优化，实现一个适合解决DataMap场景数据处理的通用处理器
 *
 * @param <T> 泛型
 * @author Y
 * @date 2023-09-18
 **/
public class DataHandler<T> {

    /**
     * 数据容器
     */
    private final DataContainer<T> dataContainer;

    /**
     * 数据消费者
     */
    private final Consumer<List<T>> dataConsumer;

    /**
     * 消费周期配置
     */
    private final ScheduleConfig scheduleConfig;

    /**
     * 消费周期间隔周期的一半 毫秒
     */
    private long periodHalfMillis;

    /**
     * 上一次消费时间
     */
    private volatile long lastConsumeTime;

    /**
     * 切换状态 true切换中 false未切换
     */
    private volatile boolean change = false;

    /**
     * 数据添加计数器 大于0表示在添加中 等于0表示当前没有添加操作
     */
    private AtomicInteger addCounter;

    /**
     * 消费者 executor
     */
    private ScheduledExecutorService dataConsumerExecutor;

    /**
     * lock
     */
    private final Object lock = new Object();

    /**
     * 构造
     * 初始化
     */
    public DataHandler(DataContainer<T> dataContainer,
                       Consumer<List<T>> dataConsumer,
                       ScheduleConfig consumeScheduleConfig) {
        this.dataContainer = dataContainer;
        this.dataConsumer = dataConsumer;
        this.scheduleConfig = consumeScheduleConfig;
    }

    /**
     * 启动
     */
    public void start() {
        this.periodHalfMillis = scheduleConfig.getUnit().toMillis(scheduleConfig.getPeriod()) / 2;
        this.lastConsumeTime = System.currentTimeMillis();
        this.addCounter = new AtomicInteger();

        this.dataConsumerExecutor = Executors.newSingleThreadScheduledExecutor();
        this.dataConsumerExecutor.scheduleAtFixedRate(this::consume, scheduleConfig.getInitialDelay(),
                scheduleConfig.getPeriod(), scheduleConfig.getUnit());
    }

    /**
     * 停止
     *
     * @param shutdownNow true立刻关闭 false关闭
     */
    public void stop(boolean shutdownNow) {
        if (shutdownNow) {
            this.dataConsumerExecutor.shutdownNow();
        } else {
            this.dataConsumerExecutor.shutdown();
        }
    }

    /**
     * put 放入数据
     *
     * @param data 数据
     */
    public void put(T data) {
        if (this.change) {
            synchronized (this.lock) {
                PrintUtil.log("阻塞");
            }
        }
        addCounter.incrementAndGet();
        boolean full = !this.dataContainer.add(data);
        addCounter.decrementAndGet();
        // 容器满了
        if (full) {
            consume();
            // 重新放入
            put(data);
        }
    }

    /**
     * 消费数据
     */
    private void consume() {
        List<T> data = null;
        boolean needConsume;
        synchronized (this.lock) {
            // 锁后才修改change
            this.change = true;
            // 是否应该消费
            needConsume = needConsume();
            if (needConsume) {
                // 等待put操作结束
                while (0 != addCounter.get()) {
                    PrintUtil.log("等待put结束自旋");
                }
                data = this.dataContainer.getAndClear();
            }
            this.change = false;
        }
        if (needConsume) {
            this.lastConsumeTime = System.currentTimeMillis();
            this.dataConsumer.accept(data);
        }
    }

    /**
     * 是否应该消费
     *
     * @return true是 false否
     */
    private boolean needConsume() {
        return System.currentTimeMillis() - this.lastConsumeTime > this.periodHalfMillis
                || this.dataContainer.overHalf();
    }
}
